#pragma once
#include "Requestor.hpp"
#include <unordered_set>
namespace bitrpc
{
    namespace client
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            using SubCallback = std::function<void(const std::string &key, const std::string &msg)>; // 对于主题消息的回调处理函数
            TopicManager(const Requestor::ptr &requestor)
                : _requestor(requestor)
            {
            }
            bool create(const BaseConnection::ptr &conn, const std::string &key)
            {
                DLOG("主题创建成功");

                return commonRequest(conn, key, TopicOptype::TOPIC_CREATE);
            }
            bool remove(const BaseConnection::ptr &conn, const std::string &key)
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_REMOVE);
            }
            bool subscribe(const BaseConnection::ptr &conn, const std::string &key, const SubCallback &cb)
            {
                // 订阅主题的同时还要注册该主题的回调处理函数，添加到映射关系中
                // 如果先发送订阅请求，可能出现刚订阅完就收到消息发布请求的情况，此时回调方法还未添加
                // 为了避免这个情况，我们先添加回调方法，再发送请求，如果订阅失败，就删除映射关系
                addSubCallback(key, cb);
                bool ret = commonRequest(conn, key, TopicOptype::TOPIC_SUBSCRIBE);
                if (!ret)
                {
                    // 订阅失败
                    delSubCallback(key);
                    return false;
                }
                DLOG("主题订阅成功");

                return true;
            }
            bool cancel(const BaseConnection::ptr &conn, const std::string &key)
            {
                delSubCallback(key);
                return  commonRequest(conn, key, TopicOptype::TOPIC_CANCEL);
            }
            bool publish(const BaseConnection::ptr &conn, const std::string &key, const std::string &msg)
            {

               bool ret= commonRequest(conn, key, TopicOptype::TOPIC_PUBLISH,msg);
               if(ret)  
               {
                DLOG("主题消息发布成功");
               }
               return ret;
             }
            // 订阅主题后，由于不知道何时会收到消息发布请求，所以要将对消息的处理采用异步进行，即放入到dispatcher模块
            // 从映射表中能找到不同主题对应的处理方法
            void onPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                DLOG("开始处理消息发布请求");
                //从msg中取出操作类型，判断是否是消息发布请求，不是则直接返回
                auto type = msg->topicOptype();
                if(type != TopicOptype::TOPIC_PUBLISH)
                {
                    ELOG("收到了错误类型的主题操作请求");
                    return;
                }

                //取出主题名称和消息内容
                std::string name = msg->TopicKey();
                std::string message =  msg->topicMsg();
                //通过主题名称，查找对应主题的回调函数，有则调用，无则报错
                auto callback = getSubCallback(name);
                if(!callback)
                {
                    ELOG("收到了 %s 主题消息，但是该消息没有主题处理回调函数",message.c_str());
                    return;
                }
                callback(name,message);
            }

        private:
            void addSubCallback(const std::string &key, const SubCallback &cb)
            {
                DLOG("添加主题回调函数");
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks[key] = cb;
            }
            void delSubCallback(const std::string &key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.erase(key);
            }
            const SubCallback getSubCallback(const std::string &key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                if (_topic_callbacks.count(key) == 0)
                    return SubCallback();
                return _topic_callbacks[key];
            }
            // 通用的请求操作
            bool commonRequest(const BaseConnection::ptr &conn, const std::string &key, const TopicOptype &optype,
                               const std::string &msg = "")
            {
                // 构造一个请求对象，并填充数据
                auto msg_req = MessageFactory::create<TopicRequest>();
                msg_req->SetId(UUID::uuid());
                msg_req->SetMessageType(MessageType::REQ_TOPIC);
                msg_req->setTopicKey(key);
                msg_req->setTopicOptype(optype);
                if (optype == TopicOptype::TOPIC_PUBLISH)
                {
                    msg_req->setTopicMsg(msg);
                }
                // 发送请求，等待响应
                BaseMessage::ptr msg_rsp;
                bool ret = _requestor->send(conn, msg_req, msg_rsp);
                // 判断请求是否成功

                if (ret == false)
                {
                    ELOG("主题操作请求失败");
                    return false;
                }

                auto rpc_rsp_msg = std::dynamic_pointer_cast<TopicResponse>(msg_rsp);
                if (!rpc_rsp_msg)
                {
                    ELOG("主题操作响应，向下类型转换失败");
                    return false;
                }
                if (rpc_rsp_msg->rcode() != RCODE::RCODE_OK)
                {
                    ELOG("主题操作请求出错，%s", errReason(rpc_rsp_msg->rcode()).c_str());
                    return false;
                }
                    ELOG("主题操作请求成功");

                return true;
            }

        private:
            std::mutex _mutex;
            // 主题名称-回调函数
            // 每当订阅一个主题，就添加一个映射关系
            // 每收到一个消息，就从映射表中找到处理方法
            std::unordered_map<std::string, SubCallback> _topic_callbacks;
            Requestor::ptr _requestor; // 通过requestor来发送请求，以及接收普通响应
        };
    }
}